package kylin.obs2hive

import kylin.po.DzdpInfo
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/7/24 10:56
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object Obs2Hive {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
        .master("local[*]")
      .appName(this.getClass.getName)
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").enableHiveSupport().getOrCreate()

    import spark.implicits._

    var sc: SparkContext = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")


    sc.hadoopConfiguration.set("hive.exec.dynamic.partition.mode","nonstrict")

    sc.setLogLevel("ERROR")

    val dzdpSourcePath= "s3a://dws-data/split/split_data/2020/dzdp/6/"

    val dzdpSource: DataFrame = spark.read.orc(dzdpSourcePath)

    dzdpSource.mapPartitions(partition=>{
      partition.map(lines=>{
        val average_consume: String = lines.getString(0)
        val categoryId: String = lines.getString(1)
        val categoryName: String = lines.getString(2)
        val crawl_date: String = lines.getString(3)
        val emotionalKeywords: String = lines.getString(4)
        val evaluates: String = lines.getString(5)
        val firstCategoryId: String = lines.getString(6)
        val flavors: String = lines.getString(7)
        val food_type: String = lines.getString(8)
        val fourthCategoryId: String = lines.getString(9)
        val goodRatePercentage: String = lines.getString(10)
        val goodUrl: String = lines.getString(11)
        val good_id: String = lines.getString(12)
        val is_tuan: String = lines.getString(13)
        val loc_info: String = lines.getString(14)
        val opening_hours: String = lines.getString(15)
        val original_cost: String = lines.getString(16)
        val phone: String = lines.getString(17)
        val platformId: String = lines.getString(18)
        val platformName: String = lines.getString(19)
        val priceText: String = lines.getString(20)
        val rootCategoryId: String = lines.getString(21)
        val rootCategoryName: String = lines.getString(22)
        val salesAmount: String = lines.getString(23)
        val secondCategoryId: String = lines.getString(24)
        val sellCount: String = lines.getString(25)
        val shopCommentCount: String = lines.getString(26)
        val shopId: String = lines.getString(27)
        val shopImages: String = lines.getString(28)
        val shopName: String = lines.getString(29)
        val shopUrl: String = lines.getString(30)
        val thirdCategoryId: String = lines.getString(31)
        val dt: String = lines.getString(32)
        val title: String = lines.getString(33)
        DzdpInfo(average_consume,categoryId,categoryName,crawl_date,emotionalKeywords,evaluates,firstCategoryId,flavors,food_type,fourthCategoryId,goodRatePercentage,goodUrl,good_id,is_tuan,loc_info,opening_hours,original_cost,phone,platformId,platformName,priceText,rootCategoryId,rootCategoryName,salesAmount,secondCategoryId,sellCount,shopCommentCount,shopId,shopImages,shopName,shopUrl,thirdCategoryId,title,dt)
      })
    }).coalesce(1).write.mode(SaveMode.Append).insertInto("cy.dw_cy_dzdp_good_info")

  }
}
